package cn.itcast.storm;

import java.io.FileWriter;
import java.util.Map;
import java.util.UUID;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

/**
 * 给商品名称添加后缀，然后将数据写入文件中
 * @author Administrator
 *
 */
public class SuffixBolt extends BaseBasicBolt {
	FileWriter fileWriter = null;
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		try{
			fileWriter = new FileWriter("/home/hadoop/"+UUID.randomUUID());
		}catch(Exception e){
			e.printStackTrace();
		}
		
	}

	@Override
	public void execute(Tuple tuple, BasicOutputCollector collector) {
		//从消息元组tuple中拿到上一个组件发送过来的数据
		String upper_word = tuple.getString(0);
		//给商品名称添加后缀
		String result = upper_word + "_suffix";
		
		try{
			fileWriter.append(result);
			fileWriter.append("\n");
			fileWriter.flush();
		}catch(Exception e){
			e.printStackTrace();
		}
	}

	//声明该组件要发送出去的tuple的字段定义
	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		
		
	}

	
}
